package com.atguigu.actual.edu0417.app.dws;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.actual.edu0417.app.BaseApp;
import com.atguigu.actual.edu0417.beans.DwsTradeCourseOrderBean;
import com.atguigu.actual.edu0417.common.EduConfig;
import com.atguigu.actual.edu0417.func.DimAsyncFunction;
import com.atguigu.actual.edu0417.func.DimMapFunction;
import com.atguigu.actual.edu0417.unils.DateFormatUtil;
import com.atguigu.actual.edu0417.unils.DimInfoUtil;
import com.atguigu.actual.edu0417.unils.DorisSinkUtil;
import com.atguigu.actual.edu0417.unils.HbaseUtil;
import com.atguigu.actual.edu0417.unils.MyKafkaUtil;
import com.atguigu.actual.edu0417.unils.RedisUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.async.AsyncFunction;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.hadoop.hbase.client.Connection;
import redis.clients.jedis.Jedis;

import java.math.BigDecimal;
import java.util.Collections;
import java.util.HashSet;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.TimeUnit;

/**
 * @author: techyang
 * @since: 2023-10-21 15:40
 * @description:
 **/
public class DwsTradeCourseOrderWindow extends BaseApp {

    public static void main(String[] args) {
        DwsTradeCourseOrderWindow dwsTradeCourseOrderWindow = new DwsTradeCourseOrderWindow();
        dwsTradeCourseOrderWindow.base(args);
    }

    @Override
    public void process(StreamExecutionEnvironment env) {
        // 从kafka读取下单数据
        DataStreamSource<String> kafkaStrDs = env.fromSource(MyKafkaUtil.getKafkaSource(EduConfig.KAFKA_DWD_TRADE_ORDER_DETAIL, "dws_trade_course_order_group")
                , WatermarkStrategy.noWatermarks(), "kafka_source");

        // 将jsonStr转换为实体类对象
        SingleOutputStreamOperator<DwsTradeCourseOrderBean> beanDs = kafkaStrDs.map(new MapFunction<String, DwsTradeCourseOrderBean>() {
            @Override
            public DwsTradeCourseOrderBean map(String jsonStr) throws Exception {
                JSONObject jsonObj = JSONObject.parseObject(jsonStr);
                String courseId = jsonObj.getString("course_id");
                String courseName = jsonObj.getString("course_name");
                String orderId = jsonObj.getString("order_id");
                String userId = jsonObj.getString("user_id");
                String originAmount = jsonObj.getString("origin_amount");
                String couponReduce = jsonObj.getString("coupon_reduce");
                String finalAmount = jsonObj.getString("final_amount");
                Set<String> orderIdSet = new HashSet<>();
                orderIdSet.add(orderId);
                Set<String> userIdSet = new HashSet<>();
                userIdSet.add(userId);
                DwsTradeCourseOrderBean orderBean =
                        DwsTradeCourseOrderBean.builder()
                                .courseId(courseId)
                                .courseName(courseName)
                                .orderIdSet(orderIdSet)
                                .userIdSet(userIdSet)
                                .originAmount(new BigDecimal(originAmount))
                                .couponReduce(new BigDecimal(couponReduce))
                                .finalAmount(new BigDecimal(finalAmount))
                                .ts(jsonObj.getLong("ts") * 1000)
                                .build();
                return orderBean;
            }
        });

        // 分配水位线,提取时间戳字段
        SingleOutputStreamOperator<DwsTradeCourseOrderBean> withWatermarkDs = beanDs.assignTimestampsAndWatermarks(
                WatermarkStrategy.<DwsTradeCourseOrderBean>forMonotonousTimestamps()
                        .withTimestampAssigner(
                                (bean, ts) -> bean.getTs()
                        )
        );

        // 按照课程ID进行分组
        KeyedStream<DwsTradeCourseOrderBean, String> keyedByCourseIdDs = withWatermarkDs.keyBy(DwsTradeCourseOrderBean::getCourseId);
        // keyedByCourseIdDs.print("keyedByCourseIdDs>>>");
        // 开窗
        WindowedStream<DwsTradeCourseOrderBean, String, TimeWindow> windowDs = keyedByCourseIdDs.window(TumblingEventTimeWindows.of(Time.seconds(5)));

        // 聚合
        SingleOutputStreamOperator<DwsTradeCourseOrderBean> reduceDs = windowDs.reduce(new ReduceFunction<DwsTradeCourseOrderBean>() {
            @Override
            public DwsTradeCourseOrderBean reduce(DwsTradeCourseOrderBean value1, DwsTradeCourseOrderBean value2) throws Exception {
                value1.getOrderIdSet().addAll(value2.getOrderIdSet());
                value1.getUserIdSet().addAll(value2.getUserIdSet());
                value1.setOriginAmount(value1.getOriginAmount().add(value2.getOriginAmount()));
                value1.setCouponReduce(value1.getCouponReduce().add(value2.getCouponReduce()));
                value1.setFinalAmount(value1.getFinalAmount().add(value2.getFinalAmount()));
                return value1;
            }
        }, new WindowFunction<DwsTradeCourseOrderBean, DwsTradeCourseOrderBean, String, TimeWindow>() {
            @Override
            public void apply(String s, TimeWindow window, Iterable<DwsTradeCourseOrderBean> input, Collector<DwsTradeCourseOrderBean> out) throws Exception {
                long start = window.getStart();
                long end = window.getEnd();
                String stt = DateFormatUtil.toYmdHms(start);
                String edt = DateFormatUtil.toYmdHms(end);
                String curDate = DateFormatUtil.toDate(start);
                for (DwsTradeCourseOrderBean orderBean : input) {
                    orderBean.setStt(stt);
                    orderBean.setEdt(edt);
                    orderBean.setCurDate(curDate);
                    orderBean.setOrderCount((long) orderBean.getOrderIdSet().size());
                    orderBean.setOrderUserCount((long) orderBean.getUserIdSet().size());
                    out.collect(orderBean);
                }
            }
        });
        // reduceDs.print("reduceDs>>>");

        // 关联科目ID 旁路缓存
        /*SingleOutputStreamOperator<DwsTradeCourseOrderBean> withSubjectIdDs = reduceDs.map(new DimMapFunction<DwsTradeCourseOrderBean>("dim_course_info") {
            @Override
            protected void join(DwsTradeCourseOrderBean bean, JSONObject dimCourseJsonObj) {
                bean.setSubjectId(dimCourseJsonObj.getString("subject_id"));
            }

            @Override
            public String getKey(DwsTradeCourseOrderBean bean) {
                return bean.getCourseId();
            }
        });*/

        // 关联科目ID 旁路缓存 + 异步IO
        SingleOutputStreamOperator<DwsTradeCourseOrderBean> withSubjectIdDs = AsyncDataStream.unorderedWait(
                reduceDs,
                new DimAsyncFunction<DwsTradeCourseOrderBean>("dim_course_info") {
                    @Override
                    public void join(DwsTradeCourseOrderBean bean, JSONObject dimCourseJsonObj) {
                        bean.setSubjectId(dimCourseJsonObj.getString("subject_id"));
                    }

                    @Override
                    public String getKey(DwsTradeCourseOrderBean bean) {
                        return bean.getCourseId();
                    }
                },
                60,
                TimeUnit.SECONDS
        );
        // withSubjectIdDs.print("withSubjectIdDs>>>");
        // 关联科目信息
        SingleOutputStreamOperator<DwsTradeCourseOrderBean> withSubjectNameDs = AsyncDataStream.unorderedWait(
                withSubjectIdDs,
                new DimAsyncFunction<DwsTradeCourseOrderBean>("dim_base_subject_info") {
                    @Override
                    public void join(DwsTradeCourseOrderBean bean, JSONObject dimCourseJsonObj) {
                        bean.setSubjectName(dimCourseJsonObj.getString("subject_name"));
                        bean.setCategoryId(dimCourseJsonObj.getString("category_id"));
                    }

                    @Override
                    public String getKey(DwsTradeCourseOrderBean bean) {
                        return bean.getSubjectId();
                    }
                },
                60,
                TimeUnit.SECONDS
        );
        // 关联类别信息
        SingleOutputStreamOperator<DwsTradeCourseOrderBean> resultDs = AsyncDataStream.unorderedWait(
                withSubjectNameDs,
                new DimAsyncFunction<DwsTradeCourseOrderBean>("dim_base_category_info") {
                    @Override
                    public void join(DwsTradeCourseOrderBean bean, JSONObject dimCourseJsonObj) {
                        bean.setCategoryName(dimCourseJsonObj.getString("category_name"));
                    }

                    @Override
                    public String getKey(DwsTradeCourseOrderBean bean) {
                        return bean.getCategoryId();
                    }
                },
                60,
                TimeUnit.SECONDS
        );
        // resultDs.print("resultDs>>>");

        // 写入doris中
        resultDs.addSink(DorisSinkUtil.getJdbcSink("insert into dws_trade_course_order_window values (?,?,?,?,?,?,?,?,?,?,?,?,?,?) "));

    }
}
